package yartiss.framework.concurrency;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

public class Consumer<E> extends  Thread{
	
	private final BlockingQueue<QueueEntry<E>> queue;
	private final 	List<IElementConsumedListener<E>> listeners = new ArrayList<IElementConsumedListener<E>>();

	public Consumer(BlockingQueue<QueueEntry<E>> queue, IElementConsumedListener<E> onConsumedListener){
		this.queue = queue;
		addElementConsumedListener(onConsumedListener);
	}
	
	@Override
	public void run(){
		E entry = null;
		while(true){
			try {
				entry = queue.take().getValue();
				if(entry != null)
					fire(entry); 
				else break;
			} 
			catch (InterruptedException e) {}
		}
		System.out.println("consumer finished");
	}
	
	public void addElementConsumedListener(IElementConsumedListener<E> l){
		listeners.add(l);
	}
	
	protected void fire(E e){
		for (IElementConsumedListener<E> l : listeners) {
			l.consume(e);
		}
	}
	
	public void shutdown(){
		 try {queue.put(new QueueEntry<E>(null));} 
		 catch (InterruptedException e) {}
	}
}
